package com.bw.medical.realtime.dwd.app;

import com.bw.medical.realtime.common.base.BaseSQL;
import com.bw.medical.realtime.common.constant.Constant;
import com.bw.medical.realtime.common.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class DwdDoctorRegister extends BaseSQL {
    public static void main(String[] args) {
        new DwdDoctorRegister().start(
                10017,
                4,
                "medical_dwd_doctor_register"
        );
    }
    @Override
    public void handle(StreamExecutionEnvironment env, StreamTableEnvironment tabEnv) {
            readOdsDb(tabEnv,"medical_dwd_doctor_register");

        Table docTable = tabEnv.sqlQuery("  select\n" +
                "`data`['id'] id,\n" +
                "`data`['create_time'] register_time,\n" +
                "`data`['birthday'] birthday,\n" +
                "`data`['consultation_fee'] consultation_fee,\n" +
                "`data`['gender'] gender_code,\n" +
                "`data`['name'] `name`,\n" +
                "`data`['specialty'] specialty_cod,\n" +
                "`data`['title'] title_cod,\n" +
                "`data`['hospital_id'] hospital_id\n" +
                "from \n" +
                "topic_db\n" +
                "where `table` = 'doctor'\n" +
                "and `type` = 'insert'");
        tabEnv.createTemporaryView("doc_table",docTable);
//        docTable.execute().print();

        tabEnv.executeSql("create table " + Constant.DWD_TRADE_DOCTOR_REGISTER + "(\n" +
                "`id` string,\n" +
                "`create_time` string,\n" +
                "`birthday` string,\n" +
                "`consultation_fee` string,\n" +
                "`gender` string,\n" +
                "`name` string,\n" +
                "`specialty` string,\n" +
                "`title` string,\n" +
                "`hospital_id` string\n" +
                ") " + SQLUtil.getKafkaSinkSQL(Constant.DWD_TRADE_DOCTOR_REGISTER));

        docTable.executeInsert(Constant.DWD_TRADE_DOCTOR_REGISTER);


    }
}
